%%--------------------------------------------------------------------
%% Copyright (c) 2022-2025 EMQ Technologies Co., Ltd. All Rights Reserved.
%%--------------------------------------------------------------------
-module(emqx_connector_schema).

-behaviour(emqx_schema_hooks).

-include_lib("typerefl/include/types.hrl").
-include_lib("hocon/include/hoconsc.hrl").
-include_lib("emqx/include/logger.hrl").

-ifdef(TEST).
-include_lib("eunit/include/eunit.hrl").
-endif.

-import(hoconsc, [mk/2, ref/2]).

-export([
    top_level_common_connector_keys/0,
    project_to_connector_resource_opts/1,
    api_ref/3
]).

-export([roots/0, fields/1, desc/1, namespace/0, tags/0]).

-export([get_response/0, put_request/0, post_request/0]).

-export([connector_type_to_bridge_types/1]).

-export([
    api_fields/3,
    common_fields/0,
    connector_values/3,
    status_and_actions_fields/0,
    type_and_name_fields/1
]).

-export([
    common_resource_opts_subfields/0,
    common_resource_opts_subfields_bin/0,
    resource_opts/0,
    resource_opts_fields/0,
    resource_opts_fields/1,
    resource_opts_ref/2,
    resource_opts_ref/3
]).

-export([ehttpc_max_inactive_sc/0]).

-export([examples/1]).

-type http_method() :: get | post | put.
-type schema_example_map() :: #{atom() => term()}.

api_ref(Module, Type, Method) ->
    {Type, ref(Module, Method)}.

examples(Method) ->
    MergeFun =
        fun(Example, Examples) ->
            maps:merge(Examples, Example)
        end,
    Fun =
        fun(Module, Examples) ->
            ConnectorExamples = erlang:apply(Module, connector_examples, [Method]),
            lists:foldl(MergeFun, Examples, ConnectorExamples)
        end,
    lists:foldl(Fun, #{}, schema_modules()).

schema_modules() ->
    ConnectorTypes = emqx_connector_info:connector_types(),
    [
        emqx_connector_info:schema_module(Type)
     || Type <- ConnectorTypes
    ].

%% @doc Return connector(v2) type from the latest connector type name.
connector_type_to_bridge_types(Type) ->
    emqx_connector_info:bridge_types(Type).

project_to_connector_resource_opts(OldResourceOpts) ->
    Subfields = common_resource_opts_subfields_bin(),
    maps:with(Subfields, OldResourceOpts).

top_level_common_connector_keys() ->
    [
        <<"enable">>,
        <<"connector">>,
        <<"resource_opts">>,
        <<"description">>,
        <<"parameters">>
    ].

%%======================================================================================
%% HOCON Schema Callbacks
%%======================================================================================

%% For HTTP APIs
get_response() ->
    api_schema("get").

put_request() ->
    api_schema("put").

post_request() ->
    api_schema("post").

api_schema(Method) ->
    InfoModSchemas = emqx_connector_info_api_schemas(Method),
    hoconsc:union(connector_api_union(InfoModSchemas)).

emqx_connector_info_api_schemas(Method) ->
    ConnectorTypes = emqx_connector_info:connector_types(),
    [
        emqx_connector_info:api_schema(Type, Method)
     || Type <- ConnectorTypes
    ].

connector_api_union(Refs) ->
    Index = maps:from_list(Refs),
    fun
        (all_union_members) ->
            maps:values(Index);
        ({value, V}) ->
            case V of
                #{<<"type">> := T} ->
                    case maps:get(T, Index, undefined) of
                        undefined ->
                            throw(#{
                                field_name => type,
                                value => T,
                                reason => <<"unknown connector type">>
                            });
                        Ref ->
                            [Ref]
                    end;
                _ ->
                    maps:values(Index)
            end
    end.

%% general config
namespace() -> "connector".

tags() ->
    [<<"Connector">>].

-dialyzer({nowarn_function, roots/0}).

roots() ->
    %% TODO: drop this clause and check
    case fields(connectors) of
        [] ->
            %% TODO: drop this clause and check
            [
                {connectors,
                    ?HOCON(
                        hoconsc:map(name, typerefl:map()),
                        #{
                            importance => ?IMPORTANCE_LOW,
                            validator => fun validator/1
                        }
                    )}
            ];
        _ ->
            [
                {connectors,
                    ?HOCON(
                        ?R_REF(connectors),
                        #{
                            importance => ?IMPORTANCE_LOW,
                            validator => fun validator/1
                        }
                    )}
            ]
    end.

fields(connectors) ->
    connector_info_fields_connectors();
fields(resource_opts) ->
    resource_opts_fields();
fields("node_status") ->
    [
        node_name(),
        {"status", mk(status(), #{})},
        {"status_reason",
            mk(binary(), #{
                required => false,
                desc => ?DESC("desc_status_reason"),
                example => <<"Connection refused">>
            })}
    ].

connector_info_fields_connectors() ->
    ConnectorTypes = emqx_connector_info:connector_types(),
    [
        emqx_connector_info:config_schema(Type)
     || Type <- ConnectorTypes
    ].

desc(connectors) ->
    ?DESC("desc_connectors");
desc("node_status") ->
    ?DESC("desc_node_status");
desc(resource_opts) ->
    ?DESC(emqx_resource_schema, "creation_opts");
desc(_) ->
    undefined.

api_fields("get_connector", Type, Fields) ->
    lists:append(
        [
            type_and_name_fields(Type),
            common_fields(),
            status_and_actions_fields(),
            Fields
        ]
    );
api_fields("post_connector", Type, Fields) ->
    lists:append(
        [
            type_and_name_fields(Type),
            common_fields(),
            Fields
        ]
    );
api_fields("put_connector", _Type, Fields) ->
    lists:append(
        [
            common_fields(),
            Fields
        ]
    ).

common_fields() ->
    [
        {enable,
            mk(boolean(), #{
                desc => ?DESC("config_enable"),
                importance => ?IMPORTANCE_NO_DOC,
                default => true
            })},
        {tags, emqx_schema:tags_schema()},
        {description, emqx_schema:description_schema()}
    ].

type_and_name_fields(ConnectorType) ->
    [
        {type, mk(ConnectorType, #{required => true, desc => ?DESC("desc_type")})},
        {name, mk(binary(), #{required => true, desc => ?DESC("desc_name")})}
    ].

status_and_actions_fields() ->
    [
        {"status", mk(status(), #{desc => ?DESC("desc_status")})},
        {"status_reason",
            mk(binary(), #{
                required => false,
                desc => ?DESC("desc_status_reason"),
                example => <<"Connection refused">>
            })},
        {"node_status",
            mk(
                hoconsc:array(ref(?MODULE, "node_status")),
                #{desc => ?DESC("desc_node_status")}
            )},
        {"actions",
            mk(
                hoconsc:array(binary()),
                #{
                    desc => ?DESC("connector_actions"),
                    example => [<<"my_action">>]
                }
            )}
    ].
resource_opts_ref(Module, RefName) ->
    resource_opts_ref(Module, RefName, undefined).

resource_opts_ref(Module, RefName, ConverterFun) ->
    Meta =
        case ConverterFun of
            undefined ->
                emqx_resource_schema:resource_opts_meta();
            _ ->
                M = emqx_resource_schema:resource_opts_meta(),
                M#{converter => ConverterFun}
        end,
    [
        {resource_opts,
            mk(
                ref(Module, RefName),
                Meta
            )}
    ].

common_resource_opts_subfields() ->
    [
        health_check_interval,
        health_check_timeout,
        start_after_created,
        start_timeout
    ].

common_resource_opts_subfields_bin() ->
    lists:map(fun atom_to_binary/1, common_resource_opts_subfields()).

resource_opts() ->
    resource_opts_ref(?MODULE, resource_opts).

resource_opts_fields() ->
    resource_opts_fields(_Overrides = []).

resource_opts_fields(Overrides) ->
    %% Note: these don't include buffer-related configurations because buffer workers are
    %% tied to the action.
    ConnectorROFields = common_resource_opts_subfields(),
    lists:filter(
        fun({Key, _Sc}) -> lists:member(Key, ConnectorROFields) end,
        emqx_resource_schema:create_opts(Overrides)
    ).

ehttpc_max_inactive_sc() ->
    {max_inactive,
        mk(
            emqx_schema:timeout_duration_ms(),
            #{
                default => <<"10s">>,
                desc => ?DESC("ehttpc_max_inactive")
            }
        )}.

-spec connector_values(http_method(), atom(), schema_example_map()) -> schema_example_map().
connector_values(Method, Type, ConnectorValues) ->
    TypeBin = atom_to_binary(Type),
    lists:foldl(
        fun(M1, M2) ->
            maps:merge(M1, M2)
        end,
        #{
            description => <<"My example ", TypeBin/binary, " connector">>
        },
        [
            ConnectorValues,
            method_values(Method, Type)
        ]
    ).

method_values(post, Type) ->
    TypeBin = atom_to_binary(Type),
    #{
        name => <<TypeBin/binary, "_connector">>,
        type => TypeBin
    };
method_values(get, Type) ->
    maps:merge(
        method_values(post, Type),
        #{
            status => <<"connected">>,
            node_status => [
                #{
                    node => <<"emqx@localhost">>,
                    status => <<"connected">>
                }
            ],
            actions => [<<"my_action">>]
        }
    );
method_values(put, _Type) ->
    #{}.

%%======================================================================================
%% Helper Functions
%%======================================================================================

node_name() ->
    {"node", mk(binary(), #{desc => ?DESC("desc_node_name"), example => "emqx@127.0.0.1"})}.

status() ->
    hoconsc:enum([connected, disconnected, connecting, inconsistent]).

validator(ConnectorsRoot) ->
    ValidatorFns = emqx_schema_hooks:list_injection_point('connectors.validators', []),
    emqx_utils:foldl_while(
        fun(Fn, _Acc) ->
            case Fn(ConnectorsRoot) of
                ok ->
                    {cont, ok};
                true ->
                    {cont, ok};
                Error ->
                    {halt, Error}
            end
        end,
        ok,
        ValidatorFns
    ).

-ifdef(TEST).
-include_lib("hocon/include/hocon_types.hrl").
schema_homogeneous_test() ->
    case
        lists:foldl(
            fun({_Name, Schema}, Bads) ->
                is_bad_schema(Schema, Bads)
            end,
            [],
            fields(connectors)
        )
    of
        [] ->
            ok;
        List ->
            throw(List)
    end.

is_bad_schema(#{type := ?MAP(_, ?R_REF(Module, TypeName))}, Bads) ->
    is_bad_schema_type(Module, TypeName, Bads);
is_bad_schema(#{type := ?MAP(_, ?UNION(Types))}, Bads) when is_list(Types) ->
    is_bad_schema_types(Types, Bads);
is_bad_schema(#{type := ?MAP(_, ?UNION(Func))}, Bads) when is_function(Func, 1) ->
    Types = Func(all_union_members),
    is_bad_schema_types(Types, Bads).

is_bad_schema_types(Types, Bads) ->
    lists:foldl(
        fun
            (?R_REF(Module, TypeName), Acc) ->
                is_bad_schema_type(Module, TypeName, Acc);
            (Type, Acc) ->
                [
                    #{
                        type => Type
                    }
                    | Acc
                ]
        end,
        Bads,
        Types
    ).

is_bad_schema_type(Module, TypeName, Bads) ->
    Fields = Module:fields(TypeName),
    ExpectedFieldNames = common_field_names(),
    MissingFields = lists:filter(
        fun(Name) -> lists:keyfind(Name, 1, Fields) =:= false end, ExpectedFieldNames
    ),
    case MissingFields of
        [] ->
            Bads;
        _ ->
            [
                #{
                    schema_module => Module,
                    type_name => TypeName,
                    missing_fields => MissingFields
                }
                | Bads
            ]
    end.

common_field_names() ->
    [
        enable, description
    ].

-endif.
